package com.db.concurrent;

import java.util.*;
import java.util.concurrent.*;

public class ConnectionPoolRealization<T> implements ConnectionPool<T> {
    static class Wrapper<T> {
        public T connection;
        long timestampMilliseconds;
    }

    private final Semaphore semaphore;

    private final Deque<Wrapper<T>> openConnectionsStored = new LinkedList<>();
    private final Map<T, Integer> connectionsInUsage = new HashMap<>(); //Due to no requirements to EQUALS methos in class T have to use count of connections for each key.

    private ScheduledExecutorService scheduler;
    private final ConnectionSource<T> source;

    private final long keepAlive;
    private final int min;

    ConnectionPoolRealization(int init,int max,long keepAlive,TimeUnit timeUnit, ConnectionSource<T> connectionSource) {
        this.keepAlive = keepAlive;
        this.min = init;
        semaphore = new Semaphore(max);
        source = connectionSource;

        for (int i = 0; i < init; ++i) {
             openConnectionsStored.add(createWrapper(connectionSource.open()));
        }

        scheduler = Executors.newScheduledThreadPool(1);
        if (keepAlive > 0) {
            scheduler.schedule(this::killAliveIfRequired, keepAlive, timeUnit);
        }
    }

    //package-visible getters for test purposes. not-thread safe
    Deque<Wrapper<T>> getOpenConnectionsStored() {
        return openConnectionsStored;
    }


    @Override
    public T get() throws InterruptedException {
        semaphore.acquire();
        return getOrCreate();
    }

    @Override
    public T tryGet() {
        return semaphore.tryAcquire()? getOrCreate() : null;
    }

    @Override
    public T tryGet(long time, TimeUnit unit) throws InterruptedException {
        return semaphore.tryAcquire(time, unit)? getOrCreate() : null;
    }

    private Wrapper<T> createWrapper(T connection) {
        Wrapper<T> wrapper = new Wrapper<>();
        wrapper.connection = connection;
        wrapper.timestampMilliseconds = System.currentTimeMillis();
        return wrapper;
    }

    private long killAliveIfRequired() {
        synchronized (openConnectionsStored) {
            long currentTime = System.currentTimeMillis();
            long toWait = keepAlive;
            while (openConnectionsStored.size() > min) {
                long diff = currentTime - openConnectionsStored.getLast().timestampMilliseconds;
                if (diff >= keepAlive) {
                    source.close(openConnectionsStored.pollLast().connection);
                } else {
                    toWait = keepAlive-diff+1;
                    break;
                }
            }
            return toWait; //toWait not really used: but may be used to sleep EXACT required time;
        }
    }

    private T getOrCreate() {
        synchronized (openConnectionsStored) {
            Wrapper<T> connectionWrapper = openConnectionsStored.pollFirst();
            T toReturn;
            if (connectionWrapper == null) {
                toReturn = source.open();
            } else {
                toReturn = connectionWrapper.connection;
            }
            connectionsInUsage.put(toReturn, connectionsInUsage.getOrDefault(toReturn, 0)+1);
            return toReturn;
        }
    }

    @Override
    public void release(T object) {
        synchronized (openConnectionsStored) {
            if (!connectionsInUsage.containsKey(object)) {
                throw new IllegalStateException("May only release connections returned by get methods");
            }
            connectionsInUsage.put(object, connectionsInUsage.get(object)-1);
            openConnectionsStored.addFirst(createWrapper(object));
            semaphore.release();
        }
    }

    public void shutdown() {
        scheduler.shutdownNow();
        synchronized (openConnectionsStored) {
            for (Wrapper<T> connectionWrapper : openConnectionsStored) {
                source.close(connectionWrapper.connection);
            }
            //still need to release connections not returned by user!
            connectionsInUsage.keySet().forEach(source::close);
        }
    }
}
